package org.springframework.boot.netty.handler;

import com.google.protobuf.MessageLite;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoop;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.netty.channelInitializer.AbstractChannelInitializer;
import org.springframework.boot.autoconfigure.netty.server.NettySocketServer;
import org.springframework.boot.netty.codec.ProtoBufCodecBean;
import org.springframework.boot.netty.codec.proto.SpeakMessage;
import org.springframework.boot.netty.converter.MessageConverter;
import org.springframework.boot.netty.converter.ProtobufMessage2JsonConverter;
import org.springframework.boot.netty.exception.ConvertException;
import org.springframework.boot.netty.handler.common.AbstractChannelInboundHandlerAdapter;
import org.springframework.boot.netty.listener.Message;
import org.springframework.boot.netty.service.UserDetailService;
import org.springframework.context.ApplicationContext;

import java.util.concurrent.TimeUnit;

/**
 * Author: huoxingzhi
 * Date: 2020/12/14
 * Email: hxz_798561819@163.com
 */
public class DefaultProtoBufClientHandler extends AbstractChannelInboundHandlerAdapter {


    private final static Logger logger = LoggerFactory.getLogger(DefaultProtoBufClientHandler.class);

    private MessageConverter messageConverter;


    public DefaultProtoBufClientHandler() {
    }


    public DefaultProtoBufClientHandler(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
        for (MessageConverter messageConverter : this.messageConverters) {
            if(messageConverter.fromSupport(MessageLite.class,String.class)){
                this.messageConverter = messageConverter;
                break;
            }
        }
    }
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 获取服务端传输过来的消息
        if(!ProtoBufCodecBean.class.isAssignableFrom(msg.getClass())){
            throw new ConvertException("DefaultProtoBufServerHandler.clas-->channelRead-->"+msg.getClass()+"can not convert com.google.protobuf.Message.class");
        }

        ProtoBufCodecBean protoBufCodecBean = (ProtoBufCodecBean) msg;
        com.google.protobuf.Message message = protoBufCodecBean.getMessage();

        if(!SpeakMessage.class.isAssignableFrom(message.getClass())){
            String jsonContent = messageConverter.fromMessage(message,String.class);
            String accessToken = protoBufCodecBean.getHeaders().getOrDefault("access_token", "");
            Message<String> serviceMsg = new Message<>();
            serviceMsg.setContent(jsonContent);
            serviceMsg.setAccessToken(accessToken);
            super.channelRead(ctx,serviceMsg);
            // 释放内存
            ReferenceCountUtil.release(msg);
        }

        //心跳消息，交给其他handler
        ctx.fireChannelRead(msg);

    }


    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        //使用过程中断线重连
        final EventLoop eventLoop = ctx.channel().eventLoop();
        AbstractChannelInitializer channelInitializer = this.applicationContext.getBean(AbstractChannelInitializer.class);

       /* eventLoop().schedule()方法运行一个新的线程，10s后在线程中继续执行initAndStart()方法。*/
        eventLoop.schedule(new Runnable() {
            @Override
            public void run() {
                try {
                    applicationContext.getBean(NettySocketServer.class).initAndStart(channelInitializer);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, 10, TimeUnit.SECONDS);
    }
}
